package org.tabularium.text.se;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInput;
import java.io.ObjectInputStream;
import java.io.ObjectOutput;
import java.io.ObjectOutputStream;
import java.sql.Blob;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.tabularium.net.downloader.Resource;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.NotImplementedException;

/**
 * Simple Map storage structure stored in a sql db. The key is a String
 * (converted to a MYSQL BLOB in the database scheme). Current implementation is
 * based on MEDIUMBLOB, so the size of values is up to 16777215. Consider
 * LONGBLOB for larger values (up to 2^32 bytes). The key length is fixed to 767
 * or 1000 bytes. Callers should catch IOException if they try to add two ids
 * that have in common the first part and, therefore, are considered the same by
 * the database. In this case, the latter id cannot be included in the db.
 * Values can be of different types in the same storer.
 * 
 * @author fabio gasparetti
 */
public class SqlRawStorer extends RawStorer {
	/**
	 * Sql implementation is based on indexes and MySql limits the length of
	 * keys to MAX_KEY_LENGTH characters. For this reason, url is truncated to
	 * MAX_KEY_LENGTH, that is, if two urls differs in two characters after the
	 * MAX_KEY_LENGTH position, they will be considered the same url. Length can
	 * be up to 1000 bytes long for MyISAM tables, and 767 bytes for InnoDB
	 * tables.
	 */
	public static final int MAX_KEY_LENGTH = 767;

	protected Connection connection;

	protected String dbname = null;

	protected static final String readLock = "LOCK TABLES map READ;";

	protected static final String writeLock = "LOCK TABLES map WRITE;";

	protected static final String unlock = "UNLOCK TABLES;";

	protected static final String mapTable = "CREATE TABLE map "
			+ "(id INT NOT NULL AUTO_INCREMENT, kkey BLOB NOT NULL, value LONGBLOB NOT NULL, "
			+ "ts BIGINT NOT NULL, PRIMARY KEY id (id), KEY kkey_idx (kkey("
			+ MAX_KEY_LENGTH + ")), KEY ts_ids(ts)) "
			+ "MAX_ROWS=100000000 AVG_ROW_LENGTH=100;";

	protected static final String getValue = "SELECT value FROM map WHERE kkey=?;";

	protected static final String getValueWTimestamp = "SELECT value FROM map WHERE kkey=? AND ts=?;";

	protected static final String addValue = "INSERT INTO map (kkey, value, ts) VALUES (?,?,?);";

	protected static final String reset = "DELETE FROM map;";

	protected static final String getKeys = "SELECT kkey FROM map;";

	protected static final String getKey = "SELECT kkey FROM map WHERE id=?;";

	protected static final String getTimestamp = "SELECT ts FROM map WHERE kkey=?;";

	protected static final String getTimestampById = "SELECT ts FROM map WHERE id=?;";

	protected static final String getIds = "SELECT id FROM map;";

	protected static final String removeValue = "DELETE FROM map WHERE kkey=?;";

	protected static final String removeValueWTimestamp = "DELETE FROM map WHERE kkey=? AND ts=?;";

	protected static final String numValues = "SELECT COUNT(*) FROM map;";

	/**
	 * Initializes the link database.
	 * <p>
	 * Calls this method after having created a new <code>SqlLinkDb</code>
	 * object.
	 * <p>
	 * An example using MySql db:
	 * <code>init("com.mysql.jdbc.Driver", "jdbc:mysql://127.0.0.1/db_name?user=root", "jdbc:mysql://127.0.0.1/mysql?user=root", "db_name", true);</code>
	 * <p>
	 * where <code>db_name</code> is the name of the link db.
	 * 
	 * @param driver
	 *            the name of the class of the db driver
	 * @param url
	 *            the url used to access the db
	 * @param creationUrl
	 *            the url used by the driver to create a new db
	 * @param dbname
	 *            the name of the db
	 * @param reset
	 *            if true the db will be erased
	 * @throws ClassNotFoundException
	 * @throws SQLException
	 */
	public void init(String driver, String url, String creationUrl,
			String dbname, boolean reset) throws ClassNotFoundException,
			SQLException {
		PreparedStatement stat;
		this.dbname = dbname;
		if ((driver == null) || (url == null)) {
			throw new IllegalArgumentException();
		}

		try {
			Class.forName(driver).newInstance();
		} catch (Exception ex) {
			// IllegalAccessException and InstantiationException exceptions
			throw new ClassNotFoundException("Unable to instantiate " + driver
					+ " class");
		}

		try {
			connection = DriverManager.getConnection(url);
		} catch (SQLException ex) {
			if (ex.getErrorCode() == 1049) { // unknown database
				reset = true;
			} else {
				throw ex;
			}
		}

		if (reset) {
			createDb(creationUrl);
		}
		stat = connection.prepareStatement("USE " + dbname + ";");
		stat.execute();
		stat.close();
	}

	/**
	 * If you want override this method, just call it and than call
	 * prepareStatement to create further tables.
	 * 
	 * @param url
	 * @throws SQLException
	 */
	protected void createDb(String url) throws SQLException {
		connection = DriverManager.getConnection(url);

		PreparedStatement stat;
		stat = connection.prepareStatement("DROP DATABASE IF EXISTS " + dbname
				+ ";");
		stat.execute();
		stat.close();
		stat = connection.prepareStatement("CREATE DATABASE " + dbname + ";");
		stat.execute();
		stat.close();
		stat = connection.prepareStatement("USE " + dbname + ";");
		stat.execute();
		stat.close();
		stat = connection.prepareStatement(mapTable);
		stat.execute();
		stat.close();
	}

	@Override
	public void store(String id, byte[] obj) throws IOException {
		store(id, obj, System.currentTimeMillis());
	}

	@Override
	public void store(Resource res) throws IOException {				
		store(res.getURL(), RawStorer.serialize(res));
	}

	@Override
	public List<byte[]> retrieve(String id) throws IOException {
		List<byte[]> ret = new ArrayList<byte[]>();
		PreparedStatement stat = null;
		try {
			ResultSet rs;
			stat = connection.prepareStatement(readLock);
			stat.execute();
			stat.close();
			stat = connection.prepareStatement(getValue);
			stat.setString(1, id);
			rs = stat.executeQuery();
			while (rs.next()) {
				Blob b = rs.getBlob(1);
				InputStream is = b.getBinaryStream();
				ret.add(IOUtils.toByteArray(is));
				is.close();
			}
			rs.close();
			stat.close();
			stat = connection.prepareStatement(unlock);
			stat.execute();
			stat.close();
		} catch (Exception ex) {
			throw new IOException("Sql exception: " + ex.toString());
		} finally {
			if (stat != null)
				try {
					stat.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		}	
		return ret;
	}

	@Override
	public List<Resource> retrieveResource(String id) throws IOException {
		List<byte[]> rets = retrieve(id);		
		List<Resource> ret = new ArrayList<Resource>();
		Iterator<byte[]> iter = rets.iterator();
		while (iter.hasNext()) {
			byte[] bb = (byte[]) iter.next();
			try {
				ret.add((Resource)RawStorer.deserialize(bb));
			} catch (ClassNotFoundException e) {
				throw new IOException("Serialization exception: " + e.toString());
			}
		}
		return ret;
	}

	@Override
	public List<Long> timestamp(String key) throws IOException {
		List<Long> ret = new ArrayList<Long>();
		PreparedStatement stat = null;
		try {
			ResultSet rs;
			stat = connection.prepareStatement(readLock);
			stat.execute();
			stat.close();
			stat = connection.prepareStatement(getTimestamp);
			stat.setString(1, key);
			rs = stat.executeQuery();
			while (rs.next()) {
				ret.add(rs.getLong(1));
			}
			rs.close();
			stat.close();
			stat = connection.prepareStatement(unlock);
			stat.execute();
			stat.close();
		} catch (Exception ex) {
			throw new IOException("Sql exception: " + ex.toString());
		} finally {
			if (stat != null)
				try {
					stat.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		}	
		return ret;
	}

	@Override
	public void remove(String id) throws IOException {
		PreparedStatement stat = null;
		try {
			stat = connection.prepareStatement(writeLock);
			stat.execute();
			stat.close();
			stat = connection.prepareStatement(removeValue);
			stat.setString(1, id);
			stat.executeUpdate();
			stat.close();
			stat = connection.prepareStatement(unlock);
			stat.execute();
			stat.close();
		} catch (Exception ex) {
			throw new IOException("Sql exception: " + ex.toString());
		} finally {
			if (stat != null)
				try {
					stat.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		}	
	}

	@Override
	public DocIterator browse() throws IOException {
		PreparedStatement stat = null;
		int ids[] = new int[0];
		try {
			ResultSet rs;
			int dim = 0;
			stat = connection.prepareStatement(readLock);
			stat.execute();
			stat.close();
			// instead of going through a dynamic list,
			// get the size of the result set and create a static array
			stat = connection.prepareStatement(numValues);
			rs = stat.executeQuery();
			if (rs.next()) {
				dim = rs.getInt(1);
			}
			rs.close();
			stat.close();
			ids = new int[dim];
			stat = connection.prepareStatement(getIds);
			rs = stat.executeQuery();
			for (int i = 0; rs.next(); i++) {
				ids[i] = rs.getInt(1);
			}
			rs.close();
			stat.close();
			stat = connection.prepareStatement(unlock);
			stat.execute();
			stat.close();
		} catch (Exception ex) {
			throw new IOException("Sql exception: " + ex.toString());
		} finally {
			if (stat != null)
				try {
					stat.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		}
		return new SqlRawStorerIterator(this, ids);
	}

	String getKeyById(int id) throws IOException {
		String key = null;
		PreparedStatement stat = null;
		try {
			ResultSet rs;
			stat = connection.prepareStatement(readLock);
			stat.execute();
			stat.close();
			stat = connection.prepareStatement(getKey);
			stat.setInt(1, id);
			rs = stat.executeQuery();
			if (rs.next()) {
				key = rs.getString(1);
			}
			rs.close();
			stat.close();
			stat = connection.prepareStatement(unlock);
			stat.execute();
			stat.close();
		} catch (Exception ex) {
			throw new IOException("Sql exception: " + ex.toString());
		} finally {
			if (stat != null)
				try {
					stat.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		}
		return key;
	}

	long getTimestampById(int id) throws IOException {
		long ts = 0;
		PreparedStatement stat = null;
		try {
			ResultSet rs;
			stat = connection.prepareStatement(readLock);
			stat.execute();
			stat.close();
			stat = connection.prepareStatement(getTimestampById);
			stat.setInt(1, id);
			rs = stat.executeQuery();
			if (rs.next()) {
				ts = rs.getLong(1);
			}
			rs.close();
			stat.close();
			stat = connection.prepareStatement(unlock);
			stat.execute();
			stat.close();
		} catch (Exception ex) {
			throw new IOException("Sql exception: " + ex.toString());
		} finally {
			if (stat != null)
				try {
					stat.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		}
		return ts;
	}

	@Override
	public int size() throws IOException {
		int n = -1;
		PreparedStatement stat = null;
		try {
			ResultSet rs;
			stat = connection.prepareStatement(readLock);
			stat.execute();
			stat.close();
			stat = connection.prepareStatement(numValues);
			rs = stat.executeQuery();
			if (rs.next()) {
				n = rs.getInt(1);
			}
			rs.close();
			stat.close();
			stat = connection.prepareStatement(unlock);
			stat.execute();
			stat.close();
		} catch (Exception ex) {
			throw new IOException("Sql exception: " + ex.toString());
		} finally {
			if (stat != null)
				try {
					stat.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		}
		return n;
		
	}

	@Override
	public void clear() throws IOException {
		PreparedStatement stat = null;
		try {
			stat = connection.prepareStatement(writeLock);
			stat.execute();
			stat.close();
			stat = connection.prepareStatement(reset);
			stat.executeUpdate();
			stat.close();
			stat = connection.prepareStatement(unlock);
			stat.execute();
			stat.close();
		} catch (Exception ex) {
			throw new IOException("Sql exception: " + ex.toString());
		} finally {
			if (stat != null)
				try {
					stat.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		}		
	}


	@Override
	public byte[] retrieve(String id, long timestamp) throws IOException {
		PreparedStatement stat = null;
		byte[] value = null;
		try {
			ResultSet rs;
			stat = connection.prepareStatement(readLock);
			stat.execute();
			stat.close();
			stat = connection.prepareStatement(getValueWTimestamp);
			stat.setString(1, id);
			stat.setLong(2, timestamp);
			rs = stat.executeQuery();
			if (rs.next()) {
				Blob b = rs.getBlob(1);
				InputStream is = b.getBinaryStream();
				value = IOUtils.toByteArray(is);
				is.close();
			}
			rs.close();
			stat.close();
			stat = connection.prepareStatement(unlock);
			stat.execute();
			stat.close();
		} catch (Exception ex) {
			throw new IOException("Sql exception: " + ex.toString());
		} finally {
			if (stat != null)
				try {
					stat.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		}		
		return value;
	}

	@Override
	public void store(String id, byte[] obj, long timestamp) throws IOException {
		PreparedStatement stat = null;
		try {
			stat = connection.prepareStatement(writeLock);
			stat.execute();
			stat.close();
			// instead of going through a dynamic list,
			// get the size of the result set and create a static array
			stat = connection.prepareStatement(addValue);
			stat.setString(1, id);
			InputStream is = new ByteArrayInputStream(obj);
			stat.setBytes(2, obj);
			stat.setLong(3, timestamp);
			stat.execute();
			stat.close();
			is.close();
			stat = connection.prepareStatement(unlock);
			stat.execute();
			stat.close();
		} catch (Exception ex) {
			throw new IOException("Sql exception: " + ex.toString());
		} finally {
			if (stat != null)
				try {
					stat.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		}
	}

	@Override
	public Resource retrieveResource(String id, long timestamp)
			throws IOException {
		byte[] bb = this.retrieve(id, timestamp);
		Resource res = null;
		try {
			res = (Resource)RawStorer.deserialize(bb);
		} catch (ClassNotFoundException e) {
			throw new IOException("Serialization exception: " + e.toString());
		}
		return res;
	}

	@Override
	public void remove(String id, long timestamp) throws IOException {
		PreparedStatement stat = null;
		try {
			stat = connection.prepareStatement(writeLock);
			stat.execute();
			stat.close();
			stat = connection.prepareStatement(this.removeValueWTimestamp);
			stat.setString(1, id);
			stat.setLong(2, timestamp);
			stat.executeUpdate();
			stat.close();
			stat = connection.prepareStatement(unlock);
			stat.execute();
			stat.close();
		} catch (Exception ex) {
			throw new IOException("Sql exception: " + ex.toString());
		} finally {
			if (stat != null)
				try {
					stat.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		}
	}

	

	public static void main(String[] args) {
//		SqlRawStorer q = null;
//		try {
//			byte[] yourBytes = null;
//			Long l;
//			q = new SqlRawStorer();
//			q.init("com.mysql.jdbc.Driver",
//					"jdbc:mysql://127.0.0.1/sqlrawstorer?user=root",
//					"jdbc:mysql://127.0.0.1/mysql?user=root", "sqlrawstorer",
//					true);
//			for (int i = 0; i < 10; i++) {
//				l = new Long((long) (Math.random() * 1000));
//				if (q.retrieve(Long.toString(l)) != null) {
//					System.out.println("skip key " + l + " already stored");
//					continue;
//				}
//				ByteArrayOutputStream bos = new ByteArrayOutputStream();
//				ObjectOutput out = null;
//				try {
//					out = new ObjectOutputStream(bos);
//					out.writeObject(l);
//					yourBytes = bos.toByteArray();
//				} finally {
//					try {
//						if (out != null) {
//							out.close();
//						}
//					} catch (IOException ex) {
//						// ignore close exception
//					}
//					try {
//						bos.close();
//					} catch (IOException ex) {
//						// ignore close exception
//					}
//				}
//
//				System.out.println("store " + l.toString() + ":" + yourBytes
//						+ " " + l.toString());
//				q.store(Long.toString(l), yourBytes);
//				System.out.println("size " + q.size());
//			}
//
//			DocIterator iter = q.browse();
//			while (iter.hasNext()) {
//				String key = (String) iter.next();
//				yourBytes = q.retrieve(key);
//				ByteArrayInputStream bis = new ByteArrayInputStream(yourBytes);
//				ObjectInput in = null;
//				try {
//					in = new ObjectInputStream(bis);
//					l = (Long) in.readObject();
//					System.out.println("read " + l + " " + yourBytes + " "
//							+ l.toString());
//
//				} finally {
//					try {
//						bis.close();
//					} catch (IOException ex) {
//						// ignore close exception
//					}
//					try {
//						if (in != null) {
//							in.close();
//						}
//					} catch (IOException ex) {
//						// ignore close exception
//					}
//				}
//			}
//
//			iter = q.browse();
//			while (iter.hasNext()) {
//				String key = (String) iter.next();
//				yourBytes = q.retrieve(key);
//				System.out.println("delete: " + key);
//				q.remove(key);
//				System.out.println("size: " + q.size());
//				ByteArrayInputStream bis = new ByteArrayInputStream(yourBytes);
//				ObjectInput in = null;
//				try {
//					in = new ObjectInputStream(bis);
//					l = (Long) in.readObject();
//					System.out.println("read " + l + " " + yourBytes + " "
//							+ l.toString());
//
//				} finally {
//					try {
//						bis.close();
//					} catch (IOException ex) {
//						// ignore close exception
//					}
//					try {
//						if (in != null) {
//							in.close();
//						}
//					} catch (IOException ex) {
//						// ignore close exception
//					}
//				}
//			}
//
//		} catch (Exception ex) {
//			System.out.println(ex.toString());
//			ex.printStackTrace();
//		}
	}
}

